package dse.messages.singleversion;

import java.util.*;

import dse.*;
import dse.messages.*;
import dse.messages.updatingshares.ActivateUpdateShareRequester;
import dse.messages.updatingshares.ActivateUpdateShareMessageHandler;

public class GetDhtDataObjectSharesTransaction 
extends Transaction
implements ShareDownloader, ActivateUpdateShareRequester, LockProcessor
{
	private DataObject desiredObject;
	private DataRequest request;
	private LinkedList<Node> residentNodes;
	private Node leader;
	private LinkedList<Node> followers;
	private int maxFailure;	
	private LinkedList<Node>  activeNodes;
	private int successCount;
	private int failureCount;
	
	private boolean failed;
	private int sharesToBeUpdate;
	private int validShareCount;
    private int requestShareCount;
	private int activateCount;
	private boolean holdLock;
	
	public GetDhtDataObjectSharesTransaction (
			Experiment experiment,
			int id,
			Node requester,
			DataObject desiredObject
			)
	{
		super(id, requester, experiment);
		
		this.requester = requester;
		this.desiredObject = desiredObject;
		this.activeNodes = null;
		this.successCount = 0;
		this.failureCount = 0;
		this.failed = false;
		//if(isUpdate){
		this.maxFailure = experiment.modules().nodeEventScheduler().getEventsPerPeriod();
		this.sharesToBeUpdate = desiredObject.shareCount();
		this.validShareCount = desiredObject.shareCount() - this.maxFailure;
		//}else{
		this.requestShareCount = desiredObject.requiredShareCount();
		//}
		this.request = null;
		this.activateCount = 0;
		this.residentNodes = null;
		this.leader = experiment.modules().assigner().assign(desiredObject, 1, experiment.network());
		this.followers = MessageUtil.getNodesByDht(experiment.network(), leader, desiredObject, experiment.modules().assigner());
		while(this.followers.size() > 3*this.maxFailure)
			this.followers.removeLast();
		if(this.leader.isDown())
			this.leader = this.followers.removeFirst();
		this.holdLock = false;
	}
	
	/**
	 * This initiates the first set of messages.
	 */
	public void start (ExperimentRecorder recorder)
	{			
		this.residentNodes = MessageUtil.getNodesByDht(experiment.network(), requester, desiredObject, experiment.modules().assigner());
		/*if(this.id == 1146){
			for(Node n : residentNodes)
				System.out.println(n.id() + ": " + n.isDown());	
		}*/
		if(isUpdate)
		{
			recorder.dataUpdateInitiated(
					super.id, requester, desiredObject, this.startTime
					);
			//recorder.debug(this, "start", time);
			request = new DataRequest(DataRequest.WRITE, requester.id(), 
					0, this.leader, this.id, desiredObject.id());
			request.setLeader(this.leader);
			this.sendLockRequest(recorder, startTime);			
		}
		else
		{
			recorder.dataRequestInitiated(
					super.id, requester, desiredObject, this.startTime
					);
			//recorder.debug(this, "start", time);
			request = new DataRequest(DataRequest.READ, requester.id(), 
				0, this.leader, this.id, desiredObject.id());
			request.setLeader(this.leader);
			//request.setSelector(this);
			this.sendLockRequest(recorder, startTime);
		}
	}
	
	private void sendLockRequest(ExperimentRecorder recorder, double time){
		requester.sequence();
		MessagePurpose purpose = this.isUpdate?MessagePurpose.DataObjectUpdate:MessagePurpose.DataObjectRequest;
		
		Message m = super.createRequestLockMessage(requester, this.leader, desiredObject, request, purpose, this);
		//recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
		
		for(int i=0; i<this.followers.size(); i++){
			Message m2 = super.createRequestLockMessage(requester, this.followers.get(i), desiredObject, request, MessagePurpose.DataObjectRequest, this);
			//recorder.debug(this, "new msg: " + m, time);
			experiment.messageQueue().enqueueMessage(m2, time);
		}
	}
	
	public void shareDownloaded (ExperimentRecorder recorder, Node from, DataObject dobj, int shareIndex, double timeLeft){
		recorder.debug(
				this, 
				"shareDownloaded{" + dobj.id() + "," + shareIndex+ "}! by " +
				requester + " in trans#" + super.id +  
				" (shareCount now = " + (/*sharesDownloaded.size()*/successCount+1)
				+ ")",
				timeLeft
				);
		
		// This should really never fail...
		assert (dobj == desiredObject);
		
		if (failed || finished)
			return;
		
		// Do something
		/*if (sharesDownloaded.contains(shareIndex)) {
			recorder.warn(
					"GetDhtDataObjectSharesTransaction #" + super.id + ": " +
					" Share #" + shareIndex + " was downloaded multiple times."
					);
			return;
		}
		
		if (finished) {
			recorder.warn(
					"GetDhtDataObjectSharesTransaction #" + super.id + ": " +
					" Share #" + shareIndex + " was downloaded AFTER the " +
					"entire data object was already successfully " +
					"reconstituted from other shares."
					);
			return;
		}*/
		
		successCount++;
		//sharesDownloaded.add(shareIndex);
		
		// Have we downloaded enough shares?
		//if (sharesDownloaded.size() >= dobj.requiredShareCount()) {
			// We are done, we have enough downloads!
		if(successCount >= this.desiredObject.requiredShareCount()){
			this.transactionSucceed(recorder, timeLeft);
			return;
		}
	}
	
	public void shareUpdated (ExperimentRecorder recorder, Node to, DataObject dobj, int shareIndex, double timeLeft)
	{
		recorder.debug(
				this, 
				"shareUpdated{" + dobj.id() + "," + shareIndex+ "}! by " +
				requester + " in trans#" + super.id +  
				" (shareCount now = " + (/*sharesDownloaded.size()*/successCount+1)
				+ ")",
				timeLeft
				);
		
		// This should really never fail...
		assert (dobj == desiredObject);
		
		if (failed || finished)
			return;
		
		// Do something
		/*if (sharesDownloaded.contains(shareIndex)) {
			recorder.warn(
					"GetDhtDataObjectSharesTransaction #" + super.id + ": " +
					" Share #" + shareIndex + " was updated multiple times."
					);
			return;
		}
		
		if (finished) {
			recorder.warn(
					"GetDhtDataObjectSharesTransaction #" + super.id + ": " +
					" Share #" + shareIndex + " was updated AFTER the " +
					"entire data object was already successfully " +
					"updated"
					);
			return;
		}
		
		sharesDownloaded.put(shareIndex, to);
		*/
		successCount++;
		if(this.activeNodes == null)
			this.activeNodes = new LinkedList<Node>();
		this.activeNodes.add(to);
		// Have we downloaded enough shares?
		//if ((sharesDownloaded.size()+this.sharesThatCouldntBeDownloaded)>=validShareCount) 
		if(successCount + failureCount >= this.sharesToBeUpdate)
		{
			// We are done, we have enough updating!
			/*finished = true;
			failed  = false;
			recorder.debug(
					"GetDataDhtObjectSharesTransaction trans#" + super.id + 
					" is finshed!!!"
					);
			recorder.dataUpdateFinished(
					super.id, 
					requester, 
					desiredObject, 
					true, 
					to,
					timeLeft
					);*/ 
			if(this.successCount >= this.validShareCount)
				AfterShareTransfer(recorder, timeLeft);
			else
				this.transactionFailed(recorder, timeLeft);
		}
	}
	
	public void shareDownloadFailed (
			ExperimentRecorder recorder, 
			DataObject dobj, 
			int shareIndex,
			double timeLeft
			)
	{
		recorder.debug(
				this, 
				"shareDownloadFailed{dobj " + dobj.id() + " shr" + 
				shareIndex + "} by " + requester + " in trans#" + super.id,
				timeLeft
				);
		// Have we already declared failure?
		if (failed || finished)
			return;
		
		if (requester.isUp() == false)
			// Then we are pretty much dead
			transactionFailed(recorder, timeLeft);
		else {
			this.requestShareCount++;
			if(this.requestShareCount > this.desiredObject.shareCount()){
				transactionFailed(recorder,timeLeft);
			}
			else{
				Message m = super.createRequestShareV2Message(this.requester, this.residentNodes.get(this.requestShareCount-1), desiredObject, 
						this.residentNodes.get(this.requestShareCount-1).getDataShare(desiredObject).getIndex(), this);
                recorder.debug(this, "new msg: " + m, timeLeft);
                experiment.messageQueue().enqueueMessage(m, timeLeft);
			}
		}
	}
	
	public void shareUpdateFailed (
			ExperimentRecorder recorder, 
			DataObject dobj, 
			int shareIndex,
			double timeLeft
			)
	{
		recorder.debug(
				this, 
				"shareUpdateFailed{dobj " + dobj.id() + " shr" + 
				shareIndex + "} by " + requester + " in trans#" + super.id,
				timeLeft
				);
		
		// Have we already declared failure?
		if (failed || finished)
			return;
		
		if (requester.isUp() == false)
			// Then we are pretty much dead
			transactionFailed(recorder, timeLeft);
		else {
			failureCount++;
			
			if(successCount + failureCount >= this.sharesToBeUpdate)
			{
				if(this.successCount >= this.validShareCount)
					AfterShareTransfer(recorder, timeLeft);
				else
					this.transactionFailed(recorder, timeLeft);
			}
			// Can we sill possibly succeed?
			/*if (sharesThatCouldntBeDownloaded >
				desiredObject.shareCount() - desiredObject.requiredShareCount())
			{
				transactionFailed(recorder, timeLeft);
				return;
			}
			else
			{
				if((sharesDownloaded.size()+this.sharesThatCouldntBeDownloaded)>=validShareCount)
				{
					//this.updateTransfer = true;
					AfterUpdateTransfer(recorder, timeLeft);
					
				}
			}*/
		}
	}

	public boolean isDoneDownloadingShares () {
		return (failed || finished);
	}
	
	public boolean isDoneUpdatingShares () {
		return (failed || finished);
	}
	
	
	private void transactionFailed (
			ExperimentRecorder recorder, 
			double timeLeft
			)
	{
		//if(finished)
			//return;
		recorder.debug(this, "transFailed", timeLeft);
		
		finished = true;
		failed = true;
		if(isUpdate)
			recorder.dataUpdateFinished(
				super.id, 
				requester, 
				desiredObject, 
				false,
				timeLeft
				);
		else{
			recorder.dataRequestFinished(
				super.id, 
				requester, 
				desiredObject, 
				false, 
				null,
				false,
				timeLeft
				);
			Message m = super.createReleaseLockMessage(this.requester, this.leader, MessagePurpose.DataObjectRequest, request, this);
			recorder.debug(this, "new msg: " + m, timeLeft);
			experiment.messageQueue().enqueueMessage(m, timeLeft);
		}
		//else{
			/*for(Node n: residentNodes){
				Message m = super.createReleaseLockMessage(this.coordinator, n, MessagePurpose.DataObjectUpdate, this);
				recorder.debug(this, "new msg: " + m, timeLeft);
				experiment.messageQueue().enqueueMessage(m, timeLeft);
			}*/
		//}
		//this.coordinator.finishCurrentDataRequest(desiredObject.id(), this.request);
		//if(this.coordinator.getRequestList(desiredObject.id())!=null && this.coordinator.getRequestList(desiredObject.id()).size()>0)
			//this.processNextRequest(this.coordinator, recorder, timeLeft);
	}
	
	private void transactionSucceed (
			ExperimentRecorder recorder, 
			double timeLeft
			)
	{
            //if(this.coordinator.id()== 566)
            //   System.out.println("DataRequest " + this.request.toString() + " is finished");
        finished = true;
        failed  = false;

        recorder.debug("GetDataDhtObjectSharesTransaction trans#" + super.id + " is finshed!!!"	);
        if(isUpdate){
        	recorder.dataUpdateFinished(super.id, requester, desiredObject, true, timeLeft);
        }
        else{
            recorder.dataRequestFinished(super.id, requester, desiredObject, true, this.residentNodes.get(this.validShareCount-1),
                    false, timeLeft);
            Message m = super.createReleaseLockMessage(this.requester, this.leader, MessagePurpose.DataObjectRequest, request, this);
			//recorder.debug(this, "new msg: " + m, timeLeft);
			experiment.messageQueue().enqueueMessage(m, timeLeft);
        }
	}
	
	public void shareActivated (
			ExperimentRecorder recorder,
			DataObject dobj,
			int shareIndex,
			Node node,
			double timeLeft
			)
	{
		activateCount++;
		if(node.getDataShare(desiredObject).getLatest().compareTo(this.request.getTM()) < 0)
			node.getDataShare(desiredObject).updateVersion(this.request.getTM());
		Message m = super.createReleaseLockMessage(node, this.leader, MessagePurpose.DataObjectUpdate, request, this);
		//recorder.debug(this, "new msg: " + m, timeLeft);
		experiment.messageQueue().enqueueMessage(m, timeLeft);
		if(activateCount == this.successCount)
			this.transactionSucceed(recorder, timeLeft);
	}
	
	public void shareCouldNotBeActivated (
			ExperimentRecorder recorder,
			DataObject dobj,
			int shareIndex,
			Node node,
			double timeLeft
			)
	{
		//node.removeUnactivatedUpdatedDataObjectShare(dobj, shareIndex);
		this.transactionFailed(recorder, timeLeft);
		Message m = super.createReleaseLockMessage(node, this.leader, MessagePurpose.DataObjectRequest, request, this);
		//recorder.debug(this, "new msg: " + m, timeLeft);
		experiment.messageQueue().enqueueMessage(m, timeLeft);
	}
	
	private void AfterShareTransfer(ExperimentRecorder recorder, double timeLeft)
	{
		if(this.activeNodes.size() >= desiredObject.requiredShareCount())
		{
			request.setLockNumber(this.activeNodes.size());
			for(Node n : this.activeNodes)
			{
				Message m = super.createMessage(MessageType.ActivateUpdateShare,
						requester, n, MessagePurpose.DataObjectUpdate,
						new ActivateUpdateShareMessageHandler(experiment, desiredObject,
								n.getDataShare(desiredObject).getIndex(), this)
						);
				recorder.debug(
						this, "new msg: " + m, timeLeft
						);
				experiment.messageQueue().enqueueMessage(m, timeLeft);
			}
		}
		else
			transactionFailed(recorder, timeLeft);
	}
		
	public void lockProcess(boolean success, Node from, ExperimentRecorder recorder, double time){
		if(this.holdLock)
			return;
		if(success){
			request.setLockNumber(request.getLockNumber()+1);
			if(request.getLockNumber() >= this.maxFailure+1){
				this.holdLock = true;
				if(this.request.getType() == DataRequest.READ){
		            for(int i=0; i<this.requestShareCount; i++){
		                Message m = super.createRequestShareV2Message(this.requester, this.residentNodes.get(i), desiredObject, this.residentNodes.get(i).getDataShare(desiredObject).getIndex(), this);
		                recorder.debug(this, "new msg: " + m, time);
		                experiment.messageQueue().enqueueMessage(m, time);
		            }
		        }
		        else{
		            for(Node n : this.residentNodes){
		                Message m = super.createMessage(
								MessageType.ShareDataTransfer, this.requester, n, desiredObject, MessagePurpose.DataObjectUpdate,
								new ShareDataTransferMessageHandler (
										experiment,
										desiredObject,
										n.getDataShare(desiredObject).getIndex(),
										this
										)
								);
		                recorder.debug(this, "new msg: " + m, time);
		                experiment.messageQueue().enqueueMessage(m, time);
		            }
				}
			}
			
		}	
	}
	
	public LinkedList<Node> getFollowers(){
		return this.followers;
	}
	
	public Node getRequester(){
		return this.requester;
	}
	
	/*private void processNextRequest(Node n, ExperimentRecorder recorder, double time){
		TreeSet<DataRequest> tdr = n.getRequestList(desiredObject.id());
		while(tdr.size()>0 && n.canLock(desiredObject.id(), tdr.first())){
			DataRequest dr = tdr.first();
			if(dr.getCoordinator().id() == n.id()){
				int srf = dr.getLockNumber();
				MessagePurpose purpose = MessagePurpose.DataObjectRequest;
				if(dr.getType() == DataRequest.WRITE)
					purpose = MessagePurpose.DataObjectUpdate;
				n.setHasRequest(desiredObject.id(), dr);
				//n.Lock(desiredObject.id(), dr);
				n.receiveReply(dr, n);
				n.sequence();
				for(int i=1; i<srf; i++){
					Message m = super.createRequestLockMessage(n, dr.getProcessor().getResidentNodes().get(i), desiredObject, dr, purpose, dr.getSelector(), dr.getProcessor());
					recorder.debug(this, "new msg: " + m, time);
					experiment.messageQueue().enqueueMessage(m, time);
				}
			}
			else{
				//n.Lock(desiredObject.id(), dr);
				n.sequence();
				MessagePurpose purpose = dr.getType()==DataRequest.READ?MessagePurpose.DataObjectRequest:MessagePurpose.DataObjectUpdate;
				Message m = super.createRequestLockResponseMessage(n, dr.getCoordinator(), true, purpose, dr.getSelector(), dr.getProcessor());
				recorder.debug(this, "new msg: " + m, time);
				experiment.messageQueue().enqueueMessage(m, time);
			}
			tdr.pollFirst();
		}
	}*/
}
